package com.sunday.canal.adapter.mongo.adapter;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.Message;
import com.sunday.canal.adapter.mongo.service.MongoSyncService;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;

/**
 * @Description: canal mongo 同步启动器
 * @Author: zsy
 * @Create: 2023/4/11 13:24
 */
@Slf4j
public class CanalMongoStarter extends BaseMongoStarter {

    public CanalMongoStarter(String destination, CanalConnector connector) {
        this.destination = destination;
        this.connector = connector;
    }

    protected void start() {
        thread = new Thread(this::process);
        thread.setName(destination);
        /**
         * setUncaughtExceptionHandler 用于获取线程运行时异常
         * 线程在执行时是不能抛出 checked 异常的，IDE 只会提示你用 try-catch 包裹起来。因此主线程无法直接获取子线程的线程信息，而每个 Thread 可以通过 setUncaughtExceptionHandler 注册一个回调接口
         */
        thread.setUncaughtExceptionHandler(handler);
        running = true;
        thread.start();
    }

    protected void stop() {
        if (!running) {
            return;
        }
        running = false;
        if (thread != null) {
            try {
                thread.join();
            } catch (InterruptedException e) {
                // ignore
            }
        }
    }

    protected void process() {
        int batchSize = 5 * 1024;
        while (running) {
            try {
                connector.connect();
                // 客户端订阅，不提交客户端filter，以服务端的filter为准
                connector.subscribe();
                log.info("## start the canal client");
                while (running) {
                    // 不指定 position 获取事件，该方法返回的条件: 尝试拿batchSize条记录，有多少取多少，不会阻塞等待
                    Message message = connector.getWithoutAck(batchSize);
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (batchId == -1 || size == 0) {
                        try {
                            Thread.sleep(100);
                        } catch (InterruptedException e) {
                        }
                    } else {
                        // 打印汇总
                        printSummary(message, batchId, size);
                        MongoSyncService.getInstances().sync(message.getEntries());
                    }

                    if (batchId != -1) {
                        connector.ack(batchId); // 提交确认
                    }
                }
            } catch (Throwable e) {
                log.error("process error!", e);
                try {
                    connector.rollback(); // 处理失败, 回滚数据
                } catch (Exception e1) {
                    // ignore
                    log.error("rollback error!", e);
                }
                try {
                    Thread.sleep(1000L);
                } catch (Exception e1) {
                    // ignore
                }
            } finally {
                connector.disconnect();
            }
        }
    }

}
